package com.matrix.im.infrastructure.persistence.repository.impl;

import com.matrix.api.im.vo.ConversationVO;
import com.matrix.api.user.UserApi;
import com.matrix.api.user.vo.UserInfoVO;
import com.matrix.common.constant.RedisConstants;
import com.matrix.common.exception.BusinessException;
import com.matrix.im.domain.model.ChatMessage;
import com.matrix.im.domain.model.Conversation;
import com.matrix.im.domain.repository.ConversationRepository;
import com.matrix.im.infrastructure.persistence.repository.ChatMessageMongoRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.dubbo.rpc.RpcException;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author 有点甜
 * @since 2025/8/2
 */
@Slf4j
@Repository
@RequiredArgsConstructor
public class ConversationRepositoryImpl implements ConversationRepository {

    private final MongoTemplate mongoTemplate;
    private final ChatMessageMongoRepository chatMessageMongoRepository;
    private final StringRedisTemplate redisTemplate;
    private final RedissonClient redissonClient;

    @DubboReference(check = false)
    private UserApi userApi;


    @Override
    public Conversation findOrCreateByUsers(Long user1Id, Long user2Id) {
        long u1 = Math.min(user1Id, user2Id);
        long u2 = Math.max(user1Id, user2Id);

        Query query = Query.query(Criteria.where("user1Id").is(u1).and("user2Id").is(u2));
        Conversation conversation = mongoTemplate.findOne(query, Conversation.class);

        if (conversation != null) {
            return conversation;
        }

        String lockKey = "lock:conv:create:" + u1 + ":" + u2;
        RLock lock = redissonClient.getLock(lockKey);
        try {
            if (!lock.tryLock(5, 10, TimeUnit.SECONDS)) {
                log.warn("创建会话时获取锁失败: {} <-> {}", u1, u2);
                throw new BusinessException("系统繁忙,请稍后重试");
            }
            
            conversation = mongoTemplate.findOne(query, Conversation.class);
            if (conversation != null) {
                return conversation;
            }
            
            log.info("会话不存在,创建新会话: {} <-> {}", u1, u2);
            return Conversation.create(u1, u2);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new BusinessException("创建会话时被中断");
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    @Override
    public void save(Conversation conversation) {
        mongoTemplate.save(conversation);
        if (!CollectionUtils.isEmpty(conversation.getMessages())) {
            conversation.getMessages().stream()
                    .filter(msg -> msg.getId() == null)
                    .forEach(chatMessageMongoRepository::save);
        }

        updateRedisCache(conversation);
    }

    private void updateRedisCache(Conversation conversation) {
        Long user1 = conversation.getUser1Id();
        Long user2 = conversation.getUser2Id();

        updateUserCacheInPipeline(user1, user2, conversation);
        updateUserCacheInPipeline(user2, user1, conversation);
    }

    private void updateUserCacheInPipeline(Long ownerId, Long friendId, Conversation conversation) {
        String convListKey = RedisConstants.buildKey(RedisConstants.IM_CONVERSATION_LIST_KEY_PREFIX, ownerId);
        String unreadKey = RedisConstants.buildKey(RedisConstants.IM_UNREAD_COUNT_KEY_PREFIX, ownerId);

        try {
            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                byte[] convListKeyBytes = convListKey.getBytes();
                byte[] unreadKeyBytes = unreadKey.getBytes();
                byte[] friendIdBytes = String.valueOf(friendId).getBytes();

                if (conversation.getLastMessageTimestamp() != null) {
                    connection.zSetCommands().zAdd(
                            convListKeyBytes,
                            (double) conversation.getLastMessageTimestamp().getTime(),
                            friendIdBytes
                    );
                }

                Integer unreadCount = conversation.getUnreadCountForUser(ownerId);
                if (unreadCount != null && unreadCount > 0) {
                    connection.hashCommands().hSet(
                            unreadKeyBytes,
                            friendIdBytes,
                            String.valueOf(unreadCount).getBytes()
                    );
                } else {
                    connection.hashCommands().hDel(unreadKeyBytes, friendIdBytes);
                }

                return null;
            });
        } catch (Exception e) {
            log.error("Failed to update user IM cache in pipeline for user {}.", ownerId, e);
        }
    }


    @Override
    public Page<ChatMessage> findChatMessages(Long userId, Long friendId, int pageNum, int pageSize) {
        Conversation conversation = findOrCreateByUsers(userId, friendId);
        PageRequest pageable = PageRequest.of(pageNum - 1, pageSize, Sort.by("timestamp").descending());
        return chatMessageMongoRepository.findByConversationId(conversation.getId(), pageable);
    }

    @Override
    public Page<ConversationVO> findConversationsByUserId(Long userId, int pageNum, int pageSize) {
        String convListKey = RedisConstants.buildKey(RedisConstants.IM_CONVERSATION_LIST_KEY_PREFIX, userId);
        String unreadKey = RedisConstants.buildKey(RedisConstants.IM_UNREAD_COUNT_KEY_PREFIX, userId);

        long start = (long) (pageNum - 1) * pageSize;
        long end = start + pageSize - 1;

        Set<String> friendIdStrs = redisTemplate.opsForZSet().reverseRange(convListKey, start, end);
        if (CollectionUtils.isEmpty(friendIdStrs)) {
            return new PageImpl<>(Collections.emptyList());
        }
        List<Long> friendIds = friendIdStrs.stream().map(Long::valueOf).toList();
        long total = redisTemplate.opsForZSet().size(convListKey);

        List<Object> unreadCounts = redisTemplate.opsForHash().multiGet(unreadKey,
                friendIds.stream().map(String::valueOf).collect(Collectors.toList()));

        List<Conversation> latestMessages = mongoTemplate.find(
                Query.query(new Criteria().orOperator(
                        Criteria.where("user1Id").is(userId).and("user2Id").in(friendIds),
                        Criteria.where("user2Id").is(userId).and("user1Id").in(friendIds)
                )), Conversation.class);
        Map<Long, Conversation> convMap = latestMessages.stream()
                .collect(Collectors.toMap(
                        c -> c.getUser1Id().equals(userId) ? c.getUser2Id() : c.getUser1Id(),
                        Function.identity()
                ));

        Map<Long, UserInfoVO> friendInfoMap = getFriendInfoMap(friendIds);

        List<ConversationVO> voList = new ArrayList<>();
        for (int i = 0; i < friendIds.size(); i++) {
            Long friendId = friendIds.get(i);
            Conversation conv = convMap.get(friendId);
            if (conv == null) {
                continue;
            }

            ConversationVO vo = new ConversationVO();
            UserInfoVO friendInfo = friendInfoMap.get(friendId);

            vo.setFriendId(friendId);
            vo.setFriendNickname(friendInfo != null ? friendInfo.getNickname() : "用户已注销");
            vo.setFriendAvatar(friendInfo != null ? friendInfo.getAvatar() : "default_avatar");
            vo.setLastMessageContent(conv.getLastMessageContent());
            vo.setLastMessageTimestamp(conv.getLastMessageTimestamp());

            Object unreadObj = unreadCounts.get(i);
            vo.setUnreadCount(unreadObj != null ? Integer.parseInt((String) unreadObj) : 0);

            voList.add(vo);
        }

        return new PageImpl<>(voList, PageRequest.of(pageNum - 1, pageSize), total);
    }

    @Override
    public long getTotalUnreadCount(Long userId) {
        String unreadKey = RedisConstants.buildKey(RedisConstants.IM_UNREAD_COUNT_KEY_PREFIX, userId);
        Map<Object, Object> unreadMap = redisTemplate.opsForHash().entries(unreadKey);

        return unreadMap.values().stream()
                .mapToLong(count -> Long.parseLong((String) count))
                .sum();
    }

    private Map<Long, UserInfoVO> getFriendInfoMap(List<Long> friendIds) {
        if (CollectionUtils.isEmpty(friendIds)) {
            return Collections.emptyMap();
        }
        try {
            List<UserInfoVO> usersResult = userApi.getUsersByIds(friendIds);
            if (usersResult != null) {
                return usersResult.stream().collect(Collectors.toMap(UserInfoVO::getUid, Function.identity()));
            }
        } catch (RpcException e) {
            log.error("批量获取用户信息失败, friendIds: {}, RPC异常: {}", friendIds, e.getMessage());
        }
        return Collections.emptyMap();
    }
}
